/**
 * 
 */
package xj.toolkit.script;

import java.io.ByteArrayInputStream;
import java.io.ByteArrayOutputStream;
import java.io.IOException;
import java.io.PrintStream;
import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.Callable;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.Future;
import java.util.concurrent.ThreadFactory;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.zip.ZipEntry;
import java.util.zip.ZipInputStream;

import javax.management.Notification;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.jmx.export.notification.NotificationPublisher;
import org.springframework.jmx.export.notification.NotificationPublisherAware;

import bsh.EvalError;
import bsh.Interpreter;
import xj.toolkit.util.ConcurrentByteArraysOutputStream;
import xj.toolkit.util.Pair;
import xj.toolkit.util.Triple;

/**
 * @author Marvin.Ma
 *
 */
public class ScriptExecutor implements NotificationPublisherAware {
	
    private static final Logger logger = 
    	LoggerFactory.getLogger(ScriptExecutor.class);
    
	private ExecutorService	runScriptExecutor = null;
	
	private ExecutorService	bshPrintExecutor = null;

	private	long			asyncPrintTimeout = 100;
	
	private	long			sequenceNumber = 0;
	
	private NotificationPublisher notificationPublisher;
	
	public void setNotificationPublisher(
			NotificationPublisher notificationPublisher) {
		this.notificationPublisher = notificationPublisher;
	}

	private void publishStringNotification(String msg) {
		Notification notification = 
			new Notification("script.execute", this, sequenceNumber++, msg);
		notificationPublisher.sendNotification(notification);
	}
	
	private static ExecutorService initSingleExecutor(final String threadName) {
		ExecutorService ret = Executors.newFixedThreadPool(1, new ThreadFactory(){

			public Thread newThread(Runnable r) {
				ClassLoader cl = Thread.currentThread().getContextClassLoader();
				if ( logger.isDebugEnabled() ) {
					logger.debug( threadName + " try create thread, current thread's classloader: {}", cl );
				}
				return new Thread(r, threadName);
			}});
    	
    	Future<?> future = ret.submit(new Runnable(){

			public void run() {
				if ( logger.isDebugEnabled() ) {
					logger.debug( threadName + " init thread call" );
				}
			}});
    	
    	try {
			future.get();
		} catch (InterruptedException e) {
			logger.error( "failed to init {} for {}", threadName, e );
		} catch (ExecutionException e) {
			logger.error( "failed to init {} for {}", threadName, e );
		}
		
		return ret;
	}
	
	public ScriptExecutor() {
    	runScriptExecutor = initSingleExecutor("ScriptExecutor: run bsh script thread");
    	bshPrintExecutor = initSingleExecutor("ScriptExecutor: publish bsh's print content thread");
	}
	
	public void close() {
		if ( null != runScriptExecutor ) {
			runScriptExecutor.shutdownNow();
			runScriptExecutor = null;
		}
		if ( null != bshPrintExecutor ) {
			bshPrintExecutor.shutdownNow();
			bshPrintExecutor = null;
		}
	}

	private void publishPrintContent(String prefix, byte[] bytes) {
		publishStringNotification( "[" + prefix + "]:" + new String(bytes) );
	}
	
	private void startAsyncPrintPublish(
			final AtomicBoolean isRunning,
			final String 		prefix, 
			final ConcurrentByteArraysOutputStream outputStream ) {
		bshPrintExecutor.submit(new Runnable() {

			public void run() {

				try {
					byte[] bytes = outputStream.poll(asyncPrintTimeout);
					if ( null != bytes ) {
						publishPrintContent(prefix, bytes);
					}
					else {
						if ( logger.isTraceEnabled() ) {
							logger.trace( "[" + prefix + "]'s sync print has !NO! content, goto next round.");
						}
					}
				}
				catch (InterruptedException e) {
					if ( logger.isInfoEnabled() ) {
						logger.info("[" + prefix + "]'s async print fetch interrupted.");
					}
				}
				finally {
					if ( isRunning.get() ) {
						startAsyncPrintPublish(isRunning, prefix, outputStream);
					}
					else {
						//	try to publish all left thing
						if ( outputStream.getByteArrayCount() > 0 ) {
							List<byte[]> contents = new ArrayList<byte[]>();
							if ( outputStream.drainTo(contents) > 0 ) {
								for ( byte[] bytes : contents ) {
									publishPrintContent(prefix, bytes);
								}
							}
						}
						if ( logger.isInfoEnabled() ) {
							logger.info( "[" + prefix + "]'s sync print publish stopped.");
						}
					}
				}
			}});
	}

	//			unziped string pair list, and unziped size
	private Pair< List<Pair<String, String>>, Long> unzipBytes(byte[] ziped) {
		List<Pair<String, String>> stringPairs = 
			new ArrayList<Pair<String, String>>();
		
		ZipInputStream in =
			new ZipInputStream(new ByteArrayInputStream(ziped));
		
		ZipEntry z;
		long unzipedSize = 0;
		
		try {
			while ( (z=in.getNextEntry() )!= null) {	
				String name = z.getName();
				ByteArrayOutputStream bos = new ByteArrayOutputStream();
				int b;
				while ( (b = in.read()) != -1 ) {
					bos.write(b);
				}
				if ( bos.size() > 0 ) {
					unzipedSize += bos.size();
					String content = bos.toString("UTF-8");
					stringPairs.add(Pair.of(name, content));
				}
				in.closeEntry();
			}
		} catch (Exception e) {
			logger.error("error in unzip bytes {}", e);
		} 
		finally {
			try {
				in.close();
			} catch (IOException e) {
				logger.error("error in close unzip bytes stream {}", e);
			}
		}
		
		if ( stringPairs.isEmpty() ) {
			return	null;
		}
		else {
			return	Pair.of(stringPairs, unzipedSize);
		}
	}
	
	private void processVars(Interpreter inter, byte[] ziped) {
		
		Pair< List<Pair<String, String>>, Long> ret = unzipBytes(ziped);
		if (null == ret ) {
			return;
		}
		for ( Pair<String, String> pair : ret.getFirst() ) {
			try {
				inter.set(pair.getFirst(), pair.getSecond());
			} catch (EvalError e) {
				logger.error("error in inter.set: {}", e);
				publishStringNotification( "error in inter.set: " + e );
			}
		}
		
		publishStringNotification(
				"vars compressed size : " + ziped.length 
				+ " / org size :" + ret.getSecond() 
				+ " / compress ratio:" + (float)ziped.length / (float)ret.getSecond()  
				+ "\r\n" );
	}
	
	private void processListVars(Interpreter inter, String varname,
			byte[] ziped) {
		Pair< List<Pair<String, String>>, Long> ret = unzipBytes(ziped);
		if (null == ret ) {
			return;
		}
		
		try {
			inter.set(varname, ret.getFirst());
		} catch (EvalError e) {
			logger.error("error in inter.set: {}", e);
			publishStringNotification( "error in inter.set: " + e );
		}
		
		publishStringNotification(
				"listvar [" + varname + "] compressed size : " + ziped.length 
				+ " / org size :" + ret.getSecond() 
				+ " / compress ratio:" + (float)ziped.length / (float)ret.getSecond()  
				+ "\r\n" );
	}

	private void processScripts(Interpreter inter, 
			ConcurrentByteArraysOutputStream outputStream, 
			byte[] ziped) {
		
		Pair< List<Pair<String, String>>, Long> ret = unzipBytes(ziped);
		if (null == ret ) {
			return;
		}
		for ( Pair<String, String> pair : ret.getFirst() ) {
			final AtomicBoolean	isRunning = new AtomicBoolean(true);
			
			try {
				startAsyncPrintPublish(isRunning, pair.getFirst(), outputStream);
				inter.eval( pair.getSecond() );
			}
			catch (EvalError e) {
				logger.error("error in inter.eval: {}", e);
				publishStringNotification( "error in inter.eval: " + e );
			}
			finally {
				isRunning.set(false);
			}
		}
		
		publishStringNotification(
				"scripts compressed size : " + ziped.length 
				+ " / org size :" + ret.getSecond() 
				+ " / compress ratio:" + (float)ziped.length / (float)ret.getSecond() 
				+ "\r\n" );
	}
	
	private String doExecuteScript(List<Triple<String, byte[], byte[]>> zipedTexts) {
		final ConcurrentByteArraysOutputStream outputStream = 
			new ConcurrentByteArraysOutputStream();
		
        Interpreter inter = new Interpreter();
        inter.setOut(new PrintStream(outputStream));
        
		try {
			for ( Triple<String, byte[], byte[]> tuple : zipedTexts ) {
				String type = tuple.getFirst();
				byte[] ziped = tuple.getSecond();
				if ( type.startsWith("listvar:")) {
					processListVars(inter, type.substring(8), ziped);
				}
				else if ( "var".equals(type) ) {
					processVars(inter, ziped);
				}
				else if ( "script".equals(type) ) {
					processScripts(inter, outputStream, ziped );
				}
			}
		} catch (Exception e) {
		}
		finally {
			//	stop async print publish process
		}
		
		return	null;
	}
	
	public String executeScript(final List<Triple<String, byte[], byte[]>> zipedTexts) 
		throws Exception {
		return runScriptExecutor.submit(new Callable<String>() {

			public String call() throws Exception {
				return doExecuteScript(zipedTexts);
			}}).get();
	}
	
	/**
	 * @return the asyncPrintTimeout
	 */
	public long getAsyncPrintTimeout() {
		return asyncPrintTimeout;
	}

	/**
	 * @param asyncPrintTimeout the asyncPrintTimeout to set
	 */
	public void setAsyncPrintTimeout(long asyncPrintTimeout) {
		this.asyncPrintTimeout = asyncPrintTimeout;
	}

}
